package com.enjoyor.soa.traffic.frame.support.amq;

import java.util.concurrent.TimeUnit;

import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.Session;

import org.apache.activemq.ActiveMQConnectionFactory;

import com.enjoyor.soa.traffic.util.helper.LoggerHelper;

/**
 * ActiveMQ 消息接收器，负责接收 AMQ 消息队列中的消息(一般为字符串)，具体使用代码示例如下：
 * <p>
 *
 * <pre>
 * String url = &quot;tcp://192.168.9.185:61616&quot;;
 * String queue = &quot;queue&quot;;
 * AMQReceiver receiver = new AMQReceiver(url, queue);
 * receiver.initialize();
 *
 * TextMessage text = (TextMessage) receiver.receiveMessage(1000);
 *
 * if (text != null) {
 * 	String message;
 * 	try {
 * 		message = text.getText();
 * 		process(message);
 * 	} catch (JMSException e) {
 * 		LOG.error(this.getClass().getSimpleName() + &quot; AMQ消息处理异常&quot;, e);
 * 	}
 * }
 * receiver.endEngine();
 * </pre>
 * <p>
 *
 * @author WY
 * @reviser WY
 * @date 2015-4-8
 */

public class AMQReceiver extends AbstractActiveMQ {
	// 消费者，消息接收者
	private MessageConsumer consumer;

	/**
	 * 生成一个消息接收器
	 *
	 * @param url
	 *            消息接收器URL
	 * @param queue
	 *            消息接收器名称
	 */
	public AMQReceiver(String url, String queue) {
		super(url, queue);
	}

	@Override
	public void initialize() {
		if (this.running()) {
			try {
				ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(getUrl());
				// 构造从工厂得到连接对象
				setConnection(connectionFactory.createConnection());
				// 启动
				getConnection().start();
				// 获取操作连接
				setSession(getConnection().createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE));
				// 获取session
				Destination destination = getSession().createQueue(getQueue());
				consumer = getSession().createConsumer(destination);
				LoggerHelper.LOG.info("AMQ消息接收引擎连接成功，正在尝试数据接收.......");
			} catch (JMSException e) {
				LoggerHelper.AMQ_LOG.warn(getQueue() + " AMQ消息接收引擎连接异常，正在尝试重连......");
				try {
					TimeUnit.SECONDS.sleep(CONNECTION_FAILED_DELAY);
				} catch (InterruptedException e1) {
					LoggerHelper.AMQ_LOG.warn(getQueue() + " AMQ消息发送引擎延时异常", e1);
				}
				initialize();
			}
		}
	}

	/**
	 * 接收一条消息，在指定的时间内尝试从消息队列获取消息， 如果超过指定时间没有收到消息则返回空（null）。
	 *
	 * @param time
	 *            消息接收时间
	 * @return 接收到的消息
	 */
	@Override
	public Message receiveMessage(int time) {
		Message message = null;

		try {
			message = consumer.receive(time);
		} catch (JMSException e) {
			LoggerHelper.LOG.error(getQueue() + " AMQ消息接收异常，正在准备关闭接收引擎......");
			endReceiver();
			LoggerHelper.LOG.info(getQueue() + " AMQ消息接收引擎已经处于关闭状态，正在准备重新启动......");
			initialize();
		}

		return message;
	}

	private void endReceiver() {
		try {
			consumer.close();
			getSession().close();
			getConnection().close();
		} catch (JMSException e) {
			LoggerHelper.LOG.error(getQueue() + " AMQ消息接收引擎关闭异常", e);
		}
	}

	@Override
	public void endEngine() {
		endReceiver();
		this.shutdown();
	}
}
